package org.unigram.oluolu.rqe.postprocessor;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.unigram.oluolu.common.FsUtil;
import org.unigram.oluolu.common.OluoluConstants;
import org.unigram.oluolu.common.OluoluLogger;
import org.unigram.oluolu.ngram.FrequentNGramExtration;
import org.unigram.oluolu.ngram.NgramConstants;
import org.unigram.oluolu.rqe.RqeConstants;
/**
 * An implementation of IPostProcessor. This object validates related
 * query pairs into.
 */
public abstract class NgramPostProcessor
    implements IPostProcessor {

    /** Configuration. */
    private Configuration conf = null;

    /** File system (hdfs or local). */
    private FileSystem fileSystem = null;

    /** Threshold: minimum query frequency stored into memory. */
    private final int minimumThreshold;

    /** Logger. */
    private OluoluLogger logger = OluoluLogger.getLogger();

    /**
     * Validate the pair of queries whether they are related or not based
     * on their frequencies.
     *
     * Warning: This methods need to be implemented in a inherited class.
     *
     * @param firstFreq frequency of first query
     * @param secondFreq requency of second query
     * @param pairFreq frequency of the collocation of first and
     *         second query in the sessions
     * @param totalFreq the number of queries in the input data set
     * @return true when they are related
     */
    protected abstract boolean validate(Long firstFreq,
            Long secondFreq, Long pairFreq, Long totalFreq);

    /**
     * Validate the pair of queries whether they are related or not based
     * on their string similarity.
     * 
     * @param firstQuery query string
     * @param secondQuery  query string
     * @return true when they are related 
     */
    protected abstract boolean validateByStrings(String firstQuery,
                String secondQuery);

    /**
     * Default Constructor.
     */
    private NgramPostProcessor() {
        // dummy (for safe)
        this.minimumThreshold =
                RqeConstants.DEFAULT_POST_FREQUENCY_THRESHOLD;
    }

    /**
     * Constructor.
     * 
     * @param jc Configuration
     */
    public NgramPostProcessor(final Configuration jc) {
        try {
            this.conf = jc;
            this.fileSystem = FileSystem.get(conf);
        } catch (IOException e) {
            e.printStackTrace();
            logger.logSevere("Opening Filetystem");
        }

        this.minimumThreshold =
         conf.getInt(RqeConstants.POST_FREQUENCY_THRESHOLD,
                         RqeConstants.DEFAULT_POST_FREQUENCY_THRESHOLD);
     }
    
    /**
     * Validate extracted related query pairs based on their
     * frequencies.
     *
     * @param inputFile input file name
     * @param rqeFile related query file name
     * @param outputFile output file name
     */
    public final void run(final String inputFile,
            final String rqeFile, final String outputFile) {
        String ngramFile = extractNGrams(inputFile, inputFile);
        Map<String, Long> queryFrequencies =
            this.loadNgramMap(ngramFile);
        this.outputValidatedPairs(rqeFile,
                    queryFrequencies, outputFile);
        try {
            FsUtil.clean(this.fileSystem, ngramFile);
        } catch (IOException e) {
            e.printStackTrace();
        }

        return;
    }
    
    /**
     * Extract the frequencies of the queries.
     *
     * @param inputFile input file name
     * @param outputPrefix output file prefix
     *
     * @return outputFile which have the frequent query Ngrams
     *          and the frequencies.
     */
    private String extractNGrams(final String inputFile,
            final String outputPrefix) {

        FrequentNGramExtration ngramExtractor =
            new FrequentNGramExtration();
        String outputFile = outputPrefix + "."
            + NgramConstants.DEFAULT_NGRAM_SUFFIX;

        try {
            FsUtil.checkPath(new Path(outputFile),
                    this.fileSystem);
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
        try {
            ngramExtractor.runNGram(inputFile,
                    outputPrefix, this.conf);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return outputFile;
    }    
    
    /**
     * Load ngram File and store into Map.
     *
     * @param rqeFile temporal rqe file before reduced
     *         in postprocessor
     * @param queryMap map containing queries and
     *         their frequencies
     * @param outputFile output file name
     */
    public final void outputValidatedPairs(
            final String rqeFile,
            final Map<String, Long> queryMap,
            final String outputFile) {
        FileStatus[] statuses = null;

        try {
            Path path = new Path(rqeFile);
            statuses = this.fileSystem.listStatus(path);
        } catch (IOException e) {
            e.printStackTrace();
        }

        OutputStream outputStream = null;
        Path outputPath = new Path(outputFile);
        try {
            FsUtil.checkPath(outputPath, conf);
            outputStream = this.fileSystem.create(outputPath,
                    true);
            int i = 0;
            for (FileStatus s : statuses) {
                if (s.isDir())  { continue; }
                Path p = s.getPath();
                InputStream inputStream = null;
                try {
                    inputStream =
                        this.fileSystem.open(p);
                    InputStreamReader reader =
                        new InputStreamReader(inputStream, "utf-8");
                    BufferedReader bufferedReader =
                        new BufferedReader(reader);
                    validateQueryPairs(queryMap, outputStream,
                            bufferedReader); // need to be implemented
                    outputStream.flush();
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    IOUtils.closeStream(inputStream);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(outputStream);            
        }


    }
    
    /**
     * Need to implement validate method to use this method.
     *
     * @param queryMap map containing queries and their frequencies
     * @param outputStream output
     * @param bufferedReader reader
     * @throws IOException -
     */
    protected void validateQueryPairs(
            final Map<String, Long> queryMap,
            final OutputStream outputStream,
            final BufferedReader bufferedReader)
            throws IOException {
        OutputStreamWriter writer = new OutputStreamWriter(outputStream);
        Long totalFreq
            = conf.getLong(OluoluConstants.OLUOLU_INPUT_RECORDS, 0);

        String line = "";
        int count = 0;
        while ((line = bufferedReader.readLine()) != null) {
            String[] tokens = line.split("\t");
            count += 1;

            if (count % 100 == 0) {
                System.out.println(".");
            } else if (count % 10 == 0) {
                System.out.print(".");
            }


            // FIXME: Adding quickfix to skip line on bad input
            //        What is the root cause of this?
            if (tokens.length < 2) {
                //logger.logSevere("Unexpected number of tab separated "
                //               + "fields in input. Skipping line: " + line);
                continue;
            }
            String queryPair = tokens[0];
            Long pairFrequency = new Long(0);
            try {
                pairFrequency = Long.parseLong(tokens[1]);
            } catch (NumberFormatException e) {
                //e.printStackTrace();
                continue;
            }
            String[] pairArray = queryPair.split(":");

            // FIXME: Adding quickfix to skip line on bad input
            //        What is the root cause of this?
            if (pairArray.length != 2) {
                //logger.logSevere("Unexpected input format -- "
                //              + "can't split query pair \"" + queryPair
                //              + "\".  Skipping line: " + line);
                continue;
            }

            String first = pairArray[0];
            String second = pairArray[1];
            Long firstFreq, secondFreq;
            if (queryMap.containsKey(first)) {
                firstFreq = queryMap.get(first);
            } else {
                firstFreq = new Long(0);
            }
            if (queryMap.containsKey(second)) {
                secondFreq = queryMap.get(second);
            } else {
                secondFreq = new Long(0);
            }

            boolean freqResult = this.validate(firstFreq, secondFreq,
                    pairFrequency, totalFreq);
            boolean simResult = this.validateByStrings(first, second);

            if (freqResult && simResult) {
                if (conf.getBoolean(RqeConstants.RQE_SHOW_SCORE, false)) {
                    writer.write(first + "\t" + second + "\t"
                            + secondFreq + "\n");
                } else {
                    writer.write(first + "\t" + second + "\n");
                }
            }
        }
        writer.flush();
        return;
    }
    
    /**
     * Load ngram File and store into Map.
     *
     * @param ngramFile file containing query ngram
     * @return Map object containing query and the frequency
     */
    private Map<String, Long> loadNgramMap(final String ngramFile) {
        Map<String, Long> rtMap = new HashMap<String, Long>();

        FileStatus[] status = null;
        Path path = new Path(ngramFile);
        try {
            status = this.fileSystem.listStatus(path);
        } catch (IOException e) {
            e.printStackTrace();
        }

        for (FileStatus s : status) {
            if (s.isDir()) { continue; }
            Path p = s.getPath();
            InputStream in = null;
            try {
                in = this.fileSystem.open(p);
                InputStreamReader reader =
                    new InputStreamReader(in);
                BufferedReader bufferedReader =
                    new BufferedReader(reader);
                appendQueries(rtMap, bufferedReader);
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                IOUtils.closeStream(in);
            }
        }
        return rtMap;
    }
    
    /**
     * Append query ngrams into assigned map.
     *
     * @param rtMap map for added queries and the frequencies
     * @param bufferedReader reader
     * @throws IOException -
     */
    private void appendQueries(final Map<String, Long> rtMap,
            final BufferedReader bufferedReader) throws IOException {
        String line;
        while ((line = bufferedReader.readLine()) != null) {
            //System.out.println("line: " + line);
            String[] tokens = line.split("\t");
            if (tokens.length != 2)  {
                //System.err.println("line: "
                //        +line+" do not have two segments."); //TODO throw
                continue;
            }

            String query = tokens[0];
            Long frequency = Long.parseLong(tokens[1]);
            //System.out.println("query:" + query + "\tfrequency:"
            //+ frequency + "\tthreshold: "+ this.minimumThreshold);

            if (frequency >= this.minimumThreshold) {
                rtMap.put(query, frequency);
            }
        }
        return;
    }

}
    